package com.heima.article.listener;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.heima.article.dto.ArticleStreamMessage;
import com.heima.article.dto.UpdateArticleMessage;
import com.heima.article.service.IHotArticleService;
import org.apache.kafka.common.protocol.types.Field;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;

import java.time.Duration;

/**
 * @Author 请不要叫我高司令
 * @Date 2022/4/22 18:54
 * @Version 1.0
 */

@EnableBinding(IHotArticleProcessor.class)
public class HotArticleListener {

    //注解绑定主题
    @StreamListener("hot_article_score_topic")
    @SendTo("hot_article_result_topic")
    public KStream<String, String> process(KStream<String, String> input) {

        // 输入的数据格式是 UpdateArticleMessage {"articleId":1471738975990321153,"type":0,"add":1}
        // 需要统计一段时间内的每一篇文章的操作数量,所以统计的key为文章的id
        // 把key设置为value中的id
        KStream<String, String> map = input.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                // value 为 {"articleId":1471738975990321153,"type":0,"add":1}
                UpdateArticleMessage updateArticleMessage = JSON.parseObject(value, UpdateArticleMessage.class);
                //从对象中获取id
                return new KeyValue<>(updateArticleMessage.getArticleId().toString(), value);
            }
        });

        //根据文章id进行分组
        KGroupedStream<String, String> groupByKey = map.groupByKey();
        //根据时间窗口进行统计
        TimeWindowedKStream<String, String> windowedBy = groupByKey.windowedBy(TimeWindows.of(Duration.ofSeconds(15)));
        //进行聚合处理
        //定义初始聚合结果，第一次接收到消息,返回null
        Initializer<String> init = new Initializer<String>() {
            @Override
            public String apply() {
                return null;
            }
        };

        //有一个聚合的方法
        Aggregator<String, String, String> agg = new Aggregator<String, String, String>() {
            @Override
            public String apply(String key, String value, String aggregate) {
                // key 就是文章的id
                // value 是 UpdateArticleMessage {"articleId":1471738975990321153,"type":0,"add":1}
                // aggregate 就是 ArticleStreamMessage，上一次处理完的结果,存放在特殊主题,存放时间区间内的聚合结果
                ArticleStreamMessage message = null;
                if (StrUtil.isEmpty(aggregate)) {
                    message = new ArticleStreamMessage();
                    message.setArticleId(Long.parseLong(key));
                    message.setView(0);
                    message.setLike(0);
                    message.setComment(0);
                    message.setCollect(0);
                } else {
                    message = JSON.parseObject(aggregate, ArticleStreamMessage.class);
                }
                // 根据当前的消息更新结果对象内容
                UpdateArticleMessage updateArticleMessage = JSON.parseObject(value, UpdateArticleMessage.class);
                switch (updateArticleMessage.getType()) {
                    case 0:
                        //阅读，将结果中的阅读量+当前的增量
                        message.setView(message.getView() + updateArticleMessage.getAdd());
                        break;
                    case 1:
                        // 点赞,将结果中的点赞量+当前的增量
                        message.setLike(message.getLike() + updateArticleMessage.getAdd());
                        break;
                    case 2:
                        // 评论,将结果中的评论量+当前的增量
                        message.setComment(message.getComment() + updateArticleMessage.getAdd());
                        break;
                    case 3:
                        // 收藏,将结果中的收藏量+当前的增量
                        message.setCollect(message.getCollect() + updateArticleMessage.getAdd());
                        break;
                }
                String result = JSON.toJSONString(message);
                return result;
            }
        };
        //有一个聚合的方法
        KTable<Windowed<String>, String> aggregate = windowedBy.aggregate(init, agg);
        //把k，v都转程String类型,因为配置文件中指定的是String的序列化，不转成String的话，会有问题
        KStream<String, String> stream = aggregate.toStream().map(new KeyValueMapper<Windowed<String>, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(Windowed<String> key, String value) {
                // 输出的数据格式是key 文章id,value是ArticleStreamMessage对象
                return new KeyValue<>(key.key(), value);
            }
        });
        // 输出的数据格式是key 文章id,value是ArticleStreamMessage对象
        //把聚合结果发到结果主题中
        return stream;


    }





















}
